"""
Case Type   : gs_checkse工具功能
Case Name   : 验证gs_checkse的A5检测项禁止将对象的所有权限授予PUBLIC角色检测与修复成功
Create At   : 2024/10/28
Owner       : lihongji
Description :
    1.将表的所有权授予public角色后检测A5项
    2.修复A5项后再次检测
    3.在数据库中查询系统表pg_class
    4.将视图的所有权授予public角色后检测A5项
    5.修复A5项后再次检测
    6.在数据库中查询系统表pg_class
    7.将模式的所有权授予public角色后检测A5项
    8.修复A5项后再次检测
    9.在数据库中查询系统表pg_namespace
    10.将函数的所有权授予public角色后检测A5项
    11.修复A5项后再次检测
    12.在数据库中查询系统表pg_proc
Expect      :
    1.赋权成功，打印告警信息
    2.修复成功，不打印告警信息
    3.系统表结果为空
    4.赋权成功，打印告警信息
    5.修复成功，不打印告警信息
    6.系统表结果为空
    7.赋权成功，打印告警信息
    8.修复成功，不打印告警信息
    9.系统表结果为空
    10.赋权成功，打印告警信息
    11.修复成功，不打印告警信息
    12.系统表结果为空
History     :
"""
import os
import time
import unittest

from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from yat.test import Node
from yat.test import macro


class GS_CHECKSE(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.primaryNode = Node('PrimaryDbUser')
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.env_path = macro.DB_ENV_PATH

    def test_gscheckse(self):
        step = ('----step1:将表的所有权授予public角色后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        grant_sql = ("drop table if exists a;"
                     "create table a(i int);"
                     "GRANT ALL ON a TO PUBLIC;")
        sql_res = self.sh_primary.execut_db_sql(grant_sql, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Disallow granting all privileges on objects to "
                      "the PUBLIC role.", check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.GRANT_SUCCESS_MSG,
                      sql_res, '执行失败' + step)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS,
                      sql_res, '执行失败' + step)

        step = '----step2:修复A5项后再次检测  expect:修复成功，不打印告警信息----'
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Disallow the PUBLIC role from having CREATE"
                         " permissions in the public schema",
                         check_res, '执行失败' + step)

        step = '----step3:在数据库中查询系统表pg_class expect:系统表结果为空----'
        self.log.info(step)
        check_sql = """SELECT relname,relacl FROM pg_class WHERE 
        (CAST(relacl AS TEXT) LIKE '%,=arwdDxt/%' OR CAST(relacl AS TEXT) 
        LIKE '=arwdDxt/%') AND (CAST(relacl AS TEXT) LIKE '%,=APmiv/%' OR 
        CAST(relacl AS TEXT) LIKE '=APmiv/%');"""
        sql_res = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.log.info(sql_res)
        self.assertIn('(0 rows)', sql_res, '执行失败' + step)

        step = ('----step4:将视图的所有权授予public角色后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        grant_sql = "GRANT ALL ON pg_roles TO PUBLIC;"
        sql_res = self.sh_primary.execut_db_sql(grant_sql, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Disallow granting all privileges on objects to "
                      "the PUBLIC role.", check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.GRANT_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = '----step5:修复A5项后再次检测  expect:修复成功，不打印告警信息----'
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Disallow the PUBLIC role from having CREATE"
                         " permissions in the public schema",
                         check_res, '执行失败' + step)

        step = '----step6:在数据库中查询系统表pg_class expect:系统表结果为空----'
        self.log.info(step)
        check_sql = """SELECT relname,relacl FROM pg_class WHERE 
        (CAST(relacl AS TEXT) LIKE '%,=arwdDxt/%' OR CAST(relacl AS TEXT) 
        LIKE '=arwdDxt/%') AND (CAST(relacl AS TEXT) LIKE '%,=APmiv/%' OR 
        CAST(relacl AS TEXT) LIKE '=APmiv/%');"""
        sql_res = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.log.info(sql_res)
        self.assertIn('(0 rows)', sql_res, '执行失败' + step)

        step = ('----step7:将模式的所有权授予public角色后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        grant_sql = "grant all on schema public to public;"
        sql_res = self.sh_primary.execut_db_sql(grant_sql, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Disallow granting all privileges on objects to "
                      "the PUBLIC role.", check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.GRANT_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = '----step8:修复A5项后再次检测  expect:修复成功，不打印告警信息----'
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Disallow the PUBLIC role from having CREATE"
                         " permissions in the public schema",
                         check_res, '执行失败' + step)

        step = '----step9:在数据库中查询系统表pg_namespace expect:系统表结果为空----'
        self.log.info(step)
        check_sql = """SELECT nspname,nspacl FROM pg_namespace 
        WHERE (CAST(nspacl AS TEXT) LIKE '%,=UC/%' OR CAST(nspacl AS TEXT) 
        LIKE '=UC/%') AND (CAST(nspacl AS TEXT) LIKE '%,=APm/%' OR 
        CAST(nspacl AS TEXT) LIKE '=APm/%');"""
        sql_res = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.log.info(sql_res)
        self.assertIn('(0 rows)', sql_res, '执行失败' + step)

        step = ('----step10:将模式的所有权授予public角色后检测A5项 '
                'expect:赋权成功，打印告警信息----')
        self.log.info(step)
        grant_sql = ("drop function if exists cube;"
                     "CREATE FUNCTION cube(x INTEGER) "
                     "RETURNS INTEGER AS \$$ "
                     "BEGIN RETURN x * x * x; "
                     "END; "
                     "\$$ LANGUAGE plpgsql;"
                     "GRANT ALL ON FUNCTION cube(x INTEGER) TO PUBLIC;")
        sql_res = self.sh_primary.execut_db_sql(grant_sql, dbname='postgres')
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Disallow granting all privileges on objects to "
                      "the PUBLIC role.", check_res, '执行失败' + step)
        self.log.info(sql_res)
        self.assertIn(self.constant.GRANT_SUCCESS_MSG,
                      sql_res, '执行失败' + step)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG,
                      sql_res, '执行失败' + step)

        step = '----step11:修复A5项后再次检测  expect:修复成功，不打印告警信息----'
        self.log.info(step)
        gs_checkse_cmd = (f'source {self.env_path};'
                          f'gs_checkse -i B5 --detail;'
                          f'gs_checkse -i A5 --detail')
        self.log.info(gs_checkse_cmd)
        check_res = self.common.get_sh_result(self.primaryNode, gs_checkse_cmd)
        self.assertIn("Setting Permission management succeed",
                      check_res, '执行失败' + step)
        self.assertNotIn("Disallow the PUBLIC role from having CREATE"
                         " permissions in the public schema",
                         check_res, '执行失败' + step)

        step = '----step12:在数据库中查询系统表pg_proc expect:系统表结果为空----'
        self.log.info(step)
        check_sql = """SELECT proname,proacl FROM pg_proc WHERE 
        (CAST(proacl AS TEXT) LIKE '%,=X/%' OR CAST(proacl AS TEXT) 
        LIKE '=X/%') AND (CAST(proacl AS TEXT) LIKE '%,=APm/%' OR 
        CAST(proacl AS TEXT) LIKE '=APm/%');"""
        sql_res = self.sh_primary.execut_db_sql(check_sql, dbname='postgres')
        self.log.info(sql_res)
        self.assertIn('(0 rows)', sql_res, '执行失败' + step)

    def tearDown(self):
        step = '----step13:清理环境 expect:成功----'
        self.log.info(step)
        revoke_sql = ("REVOKE ALL ON a FROM PUBLIC;"
                      "REVOKE ALL ON SCHEMA PUBLIC FROM PUBLIC;"
                      "REVOKE ALL ON FUNCTION cube(x INTEGER) FROM PUBLIC;"
                      "drop function if exists cube;"
                      "drop table if exists a;")
        sql_res = self.sh_primary.execut_db_sql(revoke_sql, dbname='postgres')
        self.assertEqual(3, sql_res.count(self.constant.REVOKE_SUCCESS_MSG),
                         '执行失败' + step)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败' + step)
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG, sql_res,
                      '执行失败' + step)
        self.log.info(f'----{os.path.basename(__file__)} end----')